package com.flink.conn;
import com.flink.entity.LogInfoDTO;
import com.flink.mapper.LogInfoMapper;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ibatis.session.SqlSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 *
 * @author Administrator
 * @date 2020/7/4
 */
public class SinkLogToMySql extends RichSinkFunction<List<LogInfoDTO>> {
    String table;
    String topic;
    public SinkLogToMySql(String table, String topic) {
        this.table = table;
        this.topic = topic;
    }
    private static final Logger LOG = LoggerFactory.getLogger(SinkLogToMySql.class);
    @Override
    public void invoke(List<LogInfoDTO> value, Context context) throws Exception {
        SqlSession sqlSession = MybatisSessionFactory.getSqlSessionFactory().openSession();
        LogInfoMapper logInfoMapper = sqlSession.getMapper(LogInfoMapper.class);
        try{
            //插入对应的表中
            if ("stocksLog".equals(table)) {
                logInfoMapper.saveStocksLog(value);
            } else if ("billLog".equals(table)) {
                logInfoMapper.saveBillLog(value);
            }else if ("basicLog".equals(table)) {
                logInfoMapper.saveBasicLog(value);
            }
            sqlSession.commit();
            sqlSession.close();
            LOG.info("MysqlSinkFunction commit transaction success"+table);
            LOG.info("成功插入" +topic+"--"+value.size() + "行数据");
        } catch (Throwable e){
            sqlSession.rollback();
            LOG.error("MysqlSinkFunction cause Exception,sqlSession transaction rollback...",e);
        }
    }
}
